package Test

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala._
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
 * 龚英杰：
 * 计算卡口平均时间后保存至redis中
 * key设置为 monitorAvgSpeed,value格式为 [卡口:平均速度,卡口2:平均速度...]（顺序不要求）
 */

object rw2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val redis = new FlinkJedisPoolConfig.Builder().setHost("192.168.23.62").build()

    val stream = env.fromSource(KafkaSource.builder()
    .setTopics("test")
    .setBootstrapServers("192.168.23.62:9092")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build()
    ,WatermarkStrategy.noWatermarks(),"test")

    stream.keyBy(_.split("\t")(2))
      .process(new KeyedProcessFunction[String,String,String] {
        var speedSum = 0
        var cont = 0
        var map = mutable.Map[String,Double]()
        var SpeedKey = mutable.Map[String,Int]()
        var ContKey = mutable.Map[String,Int]()
        override def processElement(i: String, context: KeyedProcessFunction[String, String, String]#Context, collector: Collector[String]): Unit = {
          val s = i.split("\t")(6).toInt
          val key = context.getCurrentKey

          if(SpeedKey.contains(key)){
            SpeedKey.update(key,SpeedKey(key)+s)
          }else{
            SpeedKey.put(key,s)
          }

          if(ContKey.contains(key)){
            ContKey.update(key,ContKey(key)+1)
          }else{
            ContKey.put(key,1)
          }

          if(SpeedKey.contains(key)&&ContKey.contains(key)){
            map.put(key,(SpeedKey(key)/ContKey(key)).toDouble)
          }

          val out = map.map{ case (k,v)=> s"$k:$v"}.mkString(",")

          collector.collect("["+out+"]")
        }
      }).addSink(new RedisSink[String](
          redis,
        new RedisMapper[String] {
          override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(RedisCommand.SET)

          override def getKeyFromData(t: String): String = "monitorAvgSpeed"

          override def getValueFromData(t: String): String = t
        }
      ))

    env.execute()
  }
}
